package com.xiaofu.live.id.generate.provider.service.impl;

import com.xiaofu.live.id.generate.provider.dao.IdBuilderMapper;
import com.xiaofu.live.id.generate.provider.dao.po.IdBuilderPO;
import com.xiaofu.live.id.generate.provider.service.IdBuilderService;
import com.xiaofu.live.id.generate.provider.service.bo.LocalSeqIdBO;
import com.xiaofu.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Service
@Slf4j
public class IdBuilderServiceImpl implements IdBuilderService, InitializingBean {

    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<>();
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<>();
    private static final int SEQ_ID = 1;
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("id-builder-thread-"+ThreadLocalRandom.current().nextInt(1000));
            return thread;
        }
    });
    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();
    private static final float UPDATE_RATE = 0.75f;

    @Autowired
    private IdBuilderMapper idBuilderMapper;

    @Override
    public Long increaseSeqId(Integer code) {
        if (code == null) {
            log.error("[increaseSeqId] id is error , id is {}", code);
            return null;
        }
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(code);
        if (localSeqIdBO == null) {
            log.error("[increaseSeqId] localSeqIdBO is error , localSeqIdBO is {}", localSeqIdBO);
            return null;
        }
        long returnId = localSeqIdBO.getCurrentNum().getAndIncrement();
        if (returnId > localSeqIdBO.getNextThreshold()) {
            log.error("[getSeqId] id is over limit , id is {}",code);
            return null;
        }
        this.refreshLocalSeqId(localSeqIdBO);
        return returnId;
    }

    /**
     * 刷新本地有序id段
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if ( localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart() > step * UPDATE_RATE ) {
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                // 异步进行同步id段操作
                log.info("开始尝试进行本地id段的同步操作");
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdBuilderPO idBuilderPO = idBuilderMapper.selectById(localSeqIdBO.getId());
                            tryUpdateMySqlRecord(idBuilderPO);
                            log.info("本地id段同步完成,id is {}", localSeqIdBO.getId());
                        } catch (Exception e) {
                            log.error("[refreshLocalSeqId] error is ",e);
                        }finally {
                            semaphoreMap.get(localSeqIdBO.getId()).release();
                        }
                    }
                });
            }
        }
    }

    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        if ((1 - UPDATE_RATE) * (localUnSeqIdBO.getNextThreshold() - localUnSeqIdBO.getCurrentStart()) > localUnSeqIdBO.getIdQueue().size() ) {
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdBuilderPO idBuilderPO = idBuilderMapper.selectById(localUnSeqIdBO.getId());
                            tryUpdateMySqlRecord(idBuilderPO);
                            log.info("本地无序id段同步完成, id is {}", localUnSeqIdBO.getId());
                        } catch (Exception e) {
                            log.error("[refreshLocalUnSeqId] error is ",e);
                        }finally {
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                        }
                    }
                });
            }

        }

    }

    @Override
    public Long increaseUnSeqId(Integer code) {
        if (code == null) {
            log.error("[increaseUnSeqId] id is error , id is {}", code);
            return null;
        }
        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(code);
        if (localUnSeqIdBO == null) {
            log.error("[increaseUnSeqId] localSeqIdBO is error , localSeqIdBO is {}", localUnSeqIdBO);
            return null;
        }
        Long resultId = localUnSeqIdBO.getIdQueue().poll();
        if (resultId == null) {
            log.error("[increaseUnSeqId] id is error , id is {}", code);
            return null;
        }
        this.refreshLocalUnSeqId(localUnSeqIdBO);
        return resultId;
    }

    @Override
    public String increaseSeqStrId(Integer code) {
        Long id = increaseSeqId(code);
        return id == null ? null : id.toString();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        List<IdBuilderPO> idBuilderPOList = idBuilderMapper.selectAll();
        for (IdBuilderPO idBuilderPO : idBuilderPOList) {
            log.info("服务器启动,抢占新的id段");
            tryUpdateMySqlRecord(idBuilderPO);
            semaphoreMap.put(idBuilderPO.getId(), new Semaphore(1));
        }
        log.info("id 初始化完成");
    }

    private void tryUpdateMySqlRecord(IdBuilderPO idBuilderPO) {

        int updateResult = idBuilderMapper.updateNewIdAndVersion(idBuilderPO.getId(), idBuilderPO.getVersion());
        if (updateResult > 0) {
            localIdBOHandler(idBuilderPO);
            return;
        }

        // 失败重试3次
        for (int i = 0; i < 3; i++) {
            idBuilderPO = idBuilderMapper.selectById(idBuilderPO.getId());
            updateResult = idBuilderMapper.updateNewIdAndVersion(idBuilderPO.getId(), idBuilderPO.getVersion());
            if (updateResult > 0) {
                localIdBOHandler(idBuilderPO);
                return;
            }
        }

        throw new RuntimeException("表id段占用失败, 竞争占用失败, id is " + idBuilderPO.getId());
    }

    private static void localIdBOHandler(IdBuilderPO idBuilderPO) {
        long currentStart = idBuilderPO.getCurrentStart();
        long nextThreshold = idBuilderPO.getNextThreshold();
        if (idBuilderPO.getIsSeq() == SEQ_ID) {
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            localSeqIdBO.setId(idBuilderPO.getId());
            AtomicLong atomicLong = new AtomicLong(currentStart);
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBO.setNextThreshold(nextThreshold);
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBOMap.put(localSeqIdBO.getId(), localSeqIdBO);
        } else {
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);
            localUnSeqIdBO.setId(idBuilderPO.getId());
            long begin = localUnSeqIdBO.getCurrentStart();
            long end = localUnSeqIdBO.getNextThreshold();
            List<Long> idList = new ArrayList<>();
            for (long i = begin; i <= end; i++) {
                idList.add(i);
            }
            // 将本地id段提前打乱,然后放入到队列中
            Collections.shuffle(idList);
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>(idList);
            localUnSeqIdBO.setIdQueue(idQueue);
            localUnSeqIdBOMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);
        }
    }

}
